package com.aspire.boc.mms.gwcmp.task;

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.math.BigDecimal;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.NoSuchElementException;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.aspire.boc.mms.gwcmp.dao.GwcmpProcessTaskMapper;
import com.aspire.boc.mms.gwcmp.dao.GwcmpTableUpdateInfoMapper;
import com.aspire.boc.mms.gwcmp.model.GwcmpProcessTask;
import com.aspire.boc.mms.gwcmp.model.GwcmpProcessTaskExample;
import com.aspire.boc.mms.gwcmp.model.GwcmpTableUpdateInfo;
import com.aspire.boc.mms.gwcmp.model.GwcmpTableUpdateInfoExample;
import com.aspire.boc.mms.gwcmp.util.Constants;
import com.aspire.boc.mms.wmf.util.ResourceManager;

/**
 * 黑名单/关键字上传任务
 * @author zhuguangzhu
 */
@Service
public class ImportShellTask {

    private static Log log = LogFactory.getLog(ImportShellTask.class);
    @Autowired
    private GwcmpProcessTaskMapper gwcmpProcessTaskMapper;
    
    @Autowired
    private GwcmpTableUpdateInfoMapper gwcmpTableUpdateInfoMapper;
    
    ResourceManager resourceManager = ResourceManager.getInstance();
    
    private LinkedBlockingQueue<GwcmpProcessTask> taskQueue = new LinkedBlockingQueue<GwcmpProcessTask>();
    
    public ImportShellTask() {

        super();
        // 启动任务队列执行线程
        startExecTaskThread();
        
    }

    private void startExecTaskThread() {

        Runnable execRunnable = new Runnable() {
            
            @Override
            public void run() {
            
                SimpleDateFormat dayFormat = new SimpleDateFormat("yyyyMMdd");
                SimpleDateFormat secFormat = new SimpleDateFormat("HH:mm:ss");
                File classpathFile = new File(this.getClass().getResource("/").getPath());
                String shellPath = classpathFile.getParentFile().getParentFile().getPath() + File.separator + "sqlloader" + File.separator;
                String absoluteFolderPath = resourceManager.getValue("task.upload.workpath");
                Runtime r = Runtime.getRuntime();
                int loop = 0;
                while(true) {
                	if(loop == 0){
                		//避免多条数据发生并发
                		loop = -1;
                		GwcmpProcessTask importLog = null;
                        try {
                            int queueSize = taskQueue.size();
                            importLog = taskQueue.take();
                            log.info(">>>>>current queue size: "+queueSize+" ,dequeue task[id="+importLog.getId()+"] successfully");
                                String id = importLog.getId();
                                String operateType = importLog.getOperateType() + "";
                                String operator = importLog.getOperator();
                                String filePath = importLog.getFilePath();
                                short taskType = importLog.getTaskType();
                                GwcmpProcessTask updateEntity = new GwcmpProcessTask();
                                updateEntity.setId(id);
                                updateEntity.setStatus(Short.valueOf(Constants.uploadStatus.INPROCESS.ordinal()+""));
                                // 处理中
                                gwcmpProcessTaskMapper.updateByPrimaryKeySelective(updateEntity);
                                String appendShell = "";
                                if(taskType == 1){
                                	//黑名单
                                   appendShell = shellPath+"importDBblacklist.sh";
		                       		//更新GWCMP_TABLE_UPDATE_INFO表
		                       		updateGwcmpTableUpdateInfo("GWCMP_BLACKLIST");
                                }else{
                                	appendShell = shellPath+"importDBkeyword.sh";
                                	updateGwcmpTableUpdateInfo("GWCMP_KEYWORD");
                                }
                                
                                List<String> params = new ArrayList<String>();
                                getAppendParams(params, appendShell, filePath, id, operateType, operator, absoluteFolderPath);
                                int execValue = execShell(r, params, getShellInfoLogPath(id, absoluteFolderPath, new Date(), dayFormat, secFormat, true));
                                if(execValue==0) {
                                    updateEntity.setStatus(Short.valueOf(Constants.uploadStatus.SUCCESS.ordinal()+""));
                                } else {
                                    updateEntity.setStatus(Short.valueOf(Constants.uploadStatus.FAIL.ordinal()+""));
                                }
                                updateEntity.setFinishTime(new Date());
                                // 处理完成
                                gwcmpProcessTaskMapper.updateByPrimaryKeySelective(updateEntity);
                        } catch(NoSuchElementException e) {
                            log.error("task queue occurs exception", e);
                        } catch(Exception e) {
                            log.error("task executing occurs exception ", e);
                        }
                        loop = 0;
                    }
                }
                }
                
        };
        new Thread(execRunnable).start();
    }

    public void importJob() {
        //File classpathFile = new File(this.getClass().getResource("/").getPath());
        //String shellPath = classpathFile.getParentFile().getParentFile().getPath()+File.separator+"bin"+File.separator;
        log.info(">>>>>ImportShellTask.importJob begin ......");
        GwcmpProcessTaskExample example = new GwcmpProcessTaskExample();
        example.setOrderByClause(" OPERATE_TIME asc");
        GwcmpProcessTaskExample.Criteria criteria = example.createCriteria();
        criteria.andStatusEqualTo(Short.valueOf(Constants.uploadStatus.PENDING.ordinal()+""));
        List<Short> taskTypeValue = new ArrayList<Short>();
        taskTypeValue.add(Short.valueOf("1"));
        taskTypeValue.add(Short.valueOf("2"));
        criteria.andTaskTypeIn(taskTypeValue);
        List<GwcmpProcessTask> gwcmpProcessTask = gwcmpProcessTaskMapper.selectByExample(example);
        int total = gwcmpProcessTask==null?0:gwcmpProcessTask.size();
        log.info(">>>>>get tasks total = " + total);
        if(total>0) {
        	GwcmpProcessTask uptTask = new GwcmpProcessTask();
            for(GwcmpProcessTask importLog : gwcmpProcessTask) {
                try {
                    int queueSize = taskQueue.size();
                    taskQueue.add(importLog);
                    uptTask.setId(importLog.getId());
                    uptTask.setStatus(Short.valueOf(Constants.uploadStatus.INLINE.ordinal()+""));
                    gwcmpProcessTaskMapper.updateByPrimaryKeySelective(uptTask);
                    log.info(">>>>>>>>>>current queue size: "+queueSize+"enqueue task[id="+importLog.getId()+"] successfully");
                } catch(Exception e) {
                    log.error(">>>>>task queue occurs exceptions", e);
                }
                
            }
        }
        
        log.info(">>>>>ImportShellTask.importJob end ......");
    }
    
    /**
     * 获取追加shell参数
     * @param params
     * @param shell
     * @param taskId
     * @param productId
     * @param experienceType
     * @param experienceBeginTime
     * @param experiencePeriod
     * @param experiencePassword
     * @param absoluteFolderPath
     * @param originalFile
     */
    private void getAppendParams(List<String> params, String appendPath, String filePath, String taskId, String operateType, String operator, String absoluteFolderPath) {
        params.clear();
        params.add("sh");
        params.add(appendPath);
        params.add(filePath);
        params.add(taskId);
        params.add(operateType);
        params.add(operator);
        params.add(absoluteFolderPath);
    }
    
    /**
     * 执行shell
     * @param r
     * @param params
     * @param shellInfoLogPath
     * @return
     */
    private int execShell(Runtime r, List<String> params, String shellInfoLogPath) {
        
        log.info(">>>>>begin to execute command['"+params+"']");
        BufferedReader logBr = null;
        BufferedWriter logWriter = null;
        try {
            Process p = r.exec(params.toArray(new String[1]));
            String line = null;
            // log info begin
            File logFile = new File(shellInfoLogPath);
            if(!logFile.exists()) {
                File logParentFile = logFile.getParentFile();
                if(logParentFile!=null && !logParentFile.exists()) {
                    logParentFile.mkdirs();
                }
                logFile.createNewFile();
            }
            logWriter = new BufferedWriter(new OutputStreamWriter(new FileOutputStream(logFile)));
            logBr = new BufferedReader(new InputStreamReader(p.getInputStream()));
            while((line=logBr.readLine()) != null) {
                logWriter.write(line);
                logWriter.write("\r\n");
            }
            
            int exitValue = p.waitFor();
            log.info(">>>>>success to execute command['"+params+"'], exitValue: "+exitValue);
            // log info end
            return exitValue;
        } catch (Exception e) {
            log.error(">>>>>fail to execute command['"+params+"'], info log file path["+shellInfoLogPath+"]", e);
        } finally {
            try {
                if(logBr!=null) {
                    logBr.close();
                }
                if(logWriter!=null) {
                    logWriter.close();
                }
            } catch (IOException e) {
                log.error(">>>>>file io stream occurs exception when closed", e);
            }
            
        }
        return -1;
    }

    private String getCurrentDateformatStr(SimpleDateFormat dateFormat, Date date) {
        
        return dateFormat.format(date);
    }
    
    /**
     * 生成shell执行日志文件路径
     * @param taskId
     * @param absoluteFolderPath
     * @param date
     * @param dayFormat
     * @param secFormat
     * @return
     */
    private String getShellInfoLogPath(String taskId, String absoluteFolderPath, Date date, SimpleDateFormat dayFormat, SimpleDateFormat secFormat, boolean isMultiple) {
        
        return absoluteFolderPath+File.separator+getCurrentDateformatStr(dayFormat, date)+File.separator+taskId+File.separator+
                "shell_"+getCurrentDateformatStr(secFormat, date)+(isMultiple?"_"+date.getTime():"")+".log";
    }
    
    public LinkedBlockingQueue<GwcmpProcessTask> getTaskQueue() {
    
        return taskQueue;
    }
    
    private void updateGwcmpTableUpdateInfo(String tableName){
    	GwcmpTableUpdateInfo updateInfo = new GwcmpTableUpdateInfo();
    	GwcmpTableUpdateInfoExample example = new GwcmpTableUpdateInfoExample();
    	GwcmpTableUpdateInfoExample.Criteria criteria = example.createCriteria();
    	criteria.andTableNameEqualTo(tableName);
    	List<GwcmpTableUpdateInfo> list = gwcmpTableUpdateInfoMapper.selectByExample(example);
    	if(list != null && list.size() > 0){
    		//更新GwcmpTableUpdateInfo表
    		updateInfo = list.get(0);
    		updateInfo.setUpdateTime(new Date());
    		gwcmpTableUpdateInfoMapper.updateByPrimaryKey(updateInfo);
    		log.info("updateGwcmpTableUpdateInfo "+ tableName +" update!");
    	} else {
    		//插入数据
    		updateInfo.setTableName(tableName);
    		updateInfo.setUpdateTime(new Date());
    		gwcmpTableUpdateInfoMapper.insertSelective(updateInfo);
    		log.info("updateGwcmpTableUpdateInfo "+ tableName +" insert!");
    	}
    }
    
}
